package udf.string;

import java.io.IOException;

import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class QueryReformulations extends EvalFunc<DataBag> {

	private BagFactory bagFactory = BagFactory.getInstance();

	public DataBag exec(Tuple input) throws IOException {

		HashSet<String> queries = new HashSet<String>();

		HashSet<String> reformulated = new HashSet<String>();

		DataBag bag = (DataBag) input.get(0);
		DataBag output = bagFactory.newDefaultBag();

		Iterator<Tuple> iter = bag.iterator();
		int query_pos = 2;
		int url_pos = 6;
		int source_pos = 7;
		int user_pos = 1;
		
		Hashtable<String, String> urls = new Hashtable<String, String>();

		for (int i = 0; i < input.size(); i++) {
			System.out.print(input.get(i) + " " + i);
			System.out.println("");
		}

		String user = "";
		String url = "";
		while (iter.hasNext()) {

			Tuple tt = iter.next();

			user = (String) tt.get(user_pos);
			url = (String) tt.get(url_pos);

			String query = (String) tt.get(query_pos);

			if (tt.size() > source_pos && tt.get(source_pos) != null
					&& !tt.get(source_pos).equals("")) {

				reformulated.add(query);
				urls.put(query, url);

			} else {

				queries.add(query);
			}
		}
		
		Iterator<String> iterA = reformulated.iterator();
		
		while (iterA.hasNext()) {

			String refo = iterA.next();
			Iterator<String> iterB = queries.iterator();

			while (iterB.hasNext()) {

				String q = iterB.next();

				Tuple t = TupleFactory.getInstance().newTuple();
				t.append(user);
				t.append(q);
				t.append(refo);
				t.append(urls.get(refo));
				
				if (areReformulations(q, refo)) {
					output.add(t);
				}

			}

		}

		return output;

	}

	/**
	 * B is a reformulation of A (B is bigger than A)
	 * 
	 * 
	 * @param a
	 * @param b
	 * @return
	 */

	public boolean areReformulations(String a, String b) {

		if (a.equals("") || b.equals(""))
			return false;

		if (a.equals(b))
			return false;

		String tokens_a[] = a.split("\\s+");
		String tokens_b[] = b.split("\\s+");

		HashSet<String> set_a = new HashSet<String>();
		HashSet<String> set_b = new HashSet<String>();

		for (int i = 0; i < tokens_a.length; i++) {
			set_a.add(tokens_a[i]);

		}

		for (int i = 0; i < tokens_b.length; i++) {

			set_b.add(tokens_b[i]);

		}

		if (set_b.containsAll(set_a) && set_b.size() > set_a.size())
			return true;

		return false;

	}

}
